package org.ala.tree.lsm.wal;

import java.io.IOException;
import java.nio.file.FileVisitResult;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.SimpleFileVisitor;
import java.nio.file.attribute.BasicFileAttributes;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.ala.tree.lsm.IMemTable;
import org.ala.tree.lsm.LSMConfig;
import org.ala.tree.lsm.memtable.MemTableOpEnums;
import org.ala.tree.lsm.tools.KryoTools;

/**
 * wal文件读写器
 * <p>	wal文件目录及格式
 * <p>	文件目录：
 * <p>	rootDir ------------------------ <DIR>
 * <p>		id	------------------------ <DIR>
 * <p>			{从0开始自增}.wal	  ------ <FILE>
 * <p>	wal文件格式:
 * <p>	WAL头:
 * <p>		操作类型：1子节	
 * <p>			1：add
 * <p>			2：update
 * <p>			3：del
 * <p>		key长度：有符号4字节
 * <p>		val长度：有符号4字节
 * <p>		key数据段
 * <p>		val数据段
 *
 * @author ala
 * @date 2025年5月25日
 */
public class Wal {
	/**
	 * 	wal头长度，9字节
	 *  [类型:1字节 key长度:4字节 val长度:4字节]
	 */
	public static final int HEAD_LEN = 9;
	
	
	/**
	 * 	配置
	 */
	protected LSMConfig config;
	/**
	 * 	当前MemTable的id
	 */
	protected String id;
	
	
	/**
	 * 写wal文件线程池
	 */
	protected ThreadPoolExecutor executor;
	protected List<WalWriterThread> threads;
//	protected ThreadPoolExecutor executor;
	
	
	public Wal(LSMConfig config, String id) {
		this.config = config;
		this.id = id;
		
		//	其实内部是通过ForkJoinPool来调度的，而且宿主线程还不能切
		//	那与其让你帮我写死，还不如我自己定义ForkJoinPool来跑
//		executor = Executors.newVirtualThreadPerTaskExecutor();
		threads = new LinkedList<>();
		executor = new ThreadPoolExecutor(
                config.walCount, config.walCount,
                180, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(1024),
                new ThreadFactory() {
                    AtomicInteger cnt = new AtomicInteger();
                    @Override
                    public Thread newThread(Runnable r) {
                        try {
                        	WalWriterThread t = new WalWriterThread(r, config.ROOT_DIR, config.WAL_DIR, id, cnt, config.walSizeByte(), config.WAL_BUFFER);
                        	threads.add(t);
                        	return t;
						} catch (IOException e) {
							throw new RuntimeException(e);
						}
                    }
                });
	}
	
	
	/**
	 * 	写入
	 */
	public <K,V> CompletableFuture<Boolean> write(int type, K k, V v) {
		byte[] key = KryoTools.serializ(k), val = KryoTools.serializ(v);
		CompletableFuture<Boolean> cf = new CompletableFuture<>();
		executor.execute(() -> {
			WalWriterThread t = (WalWriterThread) Thread.currentThread();
			try {
				t.write((byte)type, key, val);
				cf.complete(true);
			} catch (IOException e) {
				cf.completeExceptionally(e);
			}
		});
		return cf;
	}
	
	
	/**
	 * 	关闭
	 */
	public void close() {
		executor.close();
		threads.forEach(t -> {
			try {
				t.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		});
		threads.clear();
	}
	
	
	/**
	 * 	遍历wal目录下的所有文件
	 * 	@throws IOException 
	 */
	public static <V> void iterator(LSMConfig config, Class<V> vclass, IWalReadConsumer<V> consumer) throws IOException {
		_iterator(config, vclass, Path.of(String.format("%s/%s", config.ROOT_DIR, config.WAL_DIR)), consumer);
	}
	protected static <K,V> void _iterator(LSMConfig config, Class<V> vclass, Path root, IWalReadConsumer<V> consumer) throws IOException {
		//	遍历每一个子文件
		Files.walkFileTree(root, new SimpleFileVisitor<Path>() {
			public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) throws IOException {
				if (root.equals(dir)) {
					return FileVisitResult.CONTINUE;
				}
				System.out.println(dir.getFileName().toString());
				return FileVisitResult.CONTINUE;
			}
			public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
				WalReader wr = new WalReader(file);
				
				String[] ss = file.getParent().getFileName().toString().split("/");
				String id = ss[ss.length-1];
				
				int f = Integer.parseInt(file.getFileName().toString().split("\\.")[0]);
				
				wr.read(id, f, vclass, consumer);
				return FileVisitResult.CONTINUE;
			}
		});
	}
	
	
	/**
	 * 	还原回调
	 */
	public static interface IWalRevertConsumer<V> {
		/**
		 * 
		 */
		void accept(IMemTable<V> memTable);
	}
	/**
	 * 	读取回调
	 */
	public static interface IWalReadConsumer<V> {
		/**
		 * 
		 */
		void accept(String id, int file, MemTableOpEnums op, String k, V v);
	}
}
